/*
 *  Copyright 2013 Utkin Dmitry
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

#include <string>
#include <QtEndian>
#include <QDebug>
#include <QHostAddress>
#include <QTcpSocket>
#include "protoqtrpc/rpcmessage.pb.h"
#include "SocketRpcProtocol.h"

namespace protoqtrpc {

class SocketRpcProtocolPrivate
{
public:
    QTcpSocket* sock;
    quint32 packetSize;
    quint32 receivedSize;
    QByteArray buffer;

    SocketRpcProtocolPrivate(QTcpSocket* sock):
        sock(sock), packetSize(0), receivedSize(0)
    {
    }
};


SocketRpcProtocol::SocketRpcProtocol(QTcpSocket* sock):
    d_ptr(new SocketRpcProtocolPrivate(sock))
{
    connect(sock, SIGNAL(readyRead()), SLOT(readyRead()));
}

SocketRpcProtocol::~SocketRpcProtocol()
{
    delete d_ptr;
}

bool SocketRpcProtocol::sendMessage(const RpcMessage& msg)
{
    std::string data;
    if (!msg.SerializeToString(&data))
    {
        qDebug() << "Failed to serialize message";
        return false;
    }

    quint32 packetSize = static_cast<quint32>(data.size());
#if Q_BYTE_ORDER == Q_LITTLE_ENDIAN
    packetSize = qbswap(packetSize);
#endif
    // write packet length
    d_ptr->sock->write(reinterpret_cast<const char*>(&packetSize), sizeof(quint32));

    // write packet data
    qint64 len = d_ptr->sock->write(data.data(), data.size());

    if (len == -1)
    {
        qWarning() << "Unable to write RPC: " << d_ptr->sock->errorString();
        return false;
    }

    // TODO: len may be less than data.size.
    // need to check it on high load / big packets
    Q_ASSERT(len == static_cast<qint64>(data.size()));

    d_ptr->sock->flush();

#ifdef VERBOSE_DEBUG
    qDebug() << "Sent " << len << " bytes to " <<
                d_ptr->sock->peerAddress().toString();
#endif

    return true;
}

void SocketRpcProtocol::readyRead()
{
    qint64 bytesAvailable = 0;

    do
    {
        if (!d_ptr->packetSize)
        {
            // read packet size
            if (d_ptr->sock->bytesAvailable() < static_cast<int>(sizeof(quint32)))
                return;

            d_ptr->sock->read(reinterpret_cast<char*>(&d_ptr->packetSize), sizeof(quint32));
#if Q_BYTE_ORDER == Q_LITTLE_ENDIAN
            d_ptr->packetSize = qbswap(d_ptr->packetSize);
#endif
            d_ptr->receivedSize = 0;
            d_ptr->buffer.resize(d_ptr->packetSize);
        }

        qint64 readed = 0;
        qint64 bytesToRead = 0;
        // read packet data
        forever
        {
            bytesAvailable = d_ptr->sock->bytesAvailable();
            bytesToRead = qMin(static_cast<qint64>(d_ptr->packetSize - d_ptr->receivedSize),
                               bytesAvailable);
            if (!bytesToRead)
                break;

            readed = d_ptr->sock->read(d_ptr->buffer.data() + d_ptr->receivedSize, bytesToRead);
            if (readed == -1)
            {
                // this is an error!
                qWarning() << "Error while reading from socket: " << d_ptr->sock->errorString();
                return;
            }
            bytesAvailable -= readed;

            d_ptr->receivedSize += static_cast<quint32>(readed);
        }

        if (d_ptr->receivedSize == d_ptr->packetSize)
        {
            RpcMessage msg;

            if (!msg.ParseFromArray(d_ptr->buffer.constData(), d_ptr->buffer.size()))
            {
                qWarning() << "Failed to parse message";
            }
            else
            {
#ifdef VERBOSE_DEBUG
                qDebug() << "Received " << d_ptr->receivedSize << " bytes from " <<
                            d_ptr->sock->peerAddress().toString();
#endif
                emit messageReceived(msg);
            }

            d_ptr->packetSize = 0;
        }
    }
    while (bytesAvailable);
}

}
